package com.wutianyi.activemq.example;

import javax.jms.Connection;
import javax.jms.DeliveryMode;
import javax.jms.Destination;
import javax.jms.ExceptionListener;
import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.MessageConsumer;
import javax.jms.MessageProducer;
import javax.jms.Session;
import javax.jms.TextMessage;

import org.apache.activemq.ActiveMQConnectionFactory;

public class App {

	public static void main(String[] args) throws Exception {
		thread(new HelloWorldProducer(), false);
		Thread.sleep(10000);
		thread(new HelloWorldProducer(), false);
		thread(new HelloWorldConsumer(), false);
		Thread.sleep(10000);
//		thread(new HelloWorldConsumer(), false);
//		thread(new HelloWorldProducer(), false);
//		thread(new HelloWorldConsumer(), false);
//		thread(new HelloWorldProducer(), false);
//		Thread.sleep(1000);
//		thread(new HelloWorldConsumer(), false);
//		thread(new HelloWorldProducer(), false);
//		thread(new HelloWorldConsumer(), false);
//		thread(new HelloWorldConsumer(), false);
//		thread(new HelloWorldProducer(), false);
//		thread(new HelloWorldProducer(), false);
//		Thread.sleep(1000);
//		thread(new HelloWorldProducer(), false);
//		thread(new HelloWorldConsumer(), false);
//		thread(new HelloWorldConsumer(), false);
//		thread(new HelloWorldProducer(), false);
//		thread(new HelloWorldConsumer(), false);
//		thread(new HelloWorldProducer(), false);
//		thread(new HelloWorldConsumer(), false);
//		thread(new HelloWorldProducer(), false);
//		thread(new HelloWorldConsumer(), false);
//		thread(new HelloWorldConsumer(), false);
//		thread(new HelloWorldProducer(), false);
	}

	public static void thread(Runnable runnable, boolean daemon) {
		Thread brokerThread = new Thread(runnable);
		brokerThread.setDaemon(daemon);
		brokerThread.start();
	}

	/**
	 * @author wutianyi
	 * 定义轮询的方式，在一个broker不能使用的情况下，会尝试另一个broker
	 *failover:(tcp://localhost:61618,tcp://localhost:61616)
	 *
	 */
	public static class HelloWorldProducer implements Runnable {

		public void run() {
			try {
				ActiveMQConnectionFactory connectionFactory = new ActiveMQConnectionFactory(
						"failover:(tcp://localhost:61618,tcp://localhost:61616)");

				Connection connection = connectionFactory.createConnection();
				connection.start();

				Session session = connection.createSession(false,
						Session.AUTO_ACKNOWLEDGE);

				Destination destination = session.createQueue("TEST.FOO");
				
				MessageProducer producer = session.createProducer(destination);
				producer.setDeliveryMode(DeliveryMode.PERSISTENT);

				String text = "Hello world! From: "
						+ Thread.currentThread().getName() + " : "
						+ this.hashCode();
				TextMessage message = session.createTextMessage(text);

				System.out.println("Sent message: " + message.hashCode()
						+ " : " + Thread.currentThread().getName());

				producer.send(message);

				session.close();
				connection.close();
				
				System.out.println("Finish Sending!");
			} catch (Exception e) {

			}

		}

	}

	public static class HelloWorldConsumer implements Runnable,
			ExceptionListener {

		public void run() {
			try {
				ActiveMQConnectionFactory connectionFactory = new ActiveMQConnectionFactory(
						"failover:(tcp://localhost:61618,tcp://localhost:61616)");

				Connection connection = connectionFactory.createConnection();
				connection.start();
				connection.setExceptionListener(this);

				Session session = connection.createSession(false,
						Session.AUTO_ACKNOWLEDGE);

				Destination destination = session.createQueue("TEST.FOO");

				MessageConsumer consumer = session.createConsumer(destination);

				Message message = consumer.receive(1000);

				if (message instanceof TextMessage) {
					TextMessage textMessage = (TextMessage) message;
					String text = textMessage.getText();
					System.out.println("Recevied: " + text);
				} else {
					System.out.println("Recevied: " + message);
				}

				consumer.close();
				session.close();
				connection.close();
			} catch (Exception e) {

			}

		}

		public void onException(JMSException arg0) {
			System.out.println("JMS Exception occured. Shutting down client.");
		}

	}
}